package org.caw.hbase.explorer.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.caw.domain.Row;
import org.caw.domain.RowList;
import org.caw.exception.CawException;
import org.caw.exception.CawNoRecordFoundException;
import org.caw.hbase.HBaseTablePool;
import org.caw.hbase.explorer.GetManager;
import org.caw.util.ConvertUtil;

/**
 * Manager for GET operations on HBase
 * 
 * @author dino.keco
 * @version 0.1
 */
public class HBaseGetManagerImpl implements GetManager {

	private HBaseTablePool pool;

	private RowList executeGets(String tableName, List<String> rowkeys, List<String> columns, Integer maxversions, Long timestamp) throws CawException {
		HTableInterface table = null;
		try {
			table = pool.getTable(tableName);
			List<Get> gets = new ArrayList<Get>();

			for (String key : rowkeys) {
				gets.add(ConvertUtil.buildGet(key, columns, maxversions, timestamp));
			}

			Result[] results = table.get(gets);
			if (results.length == 0) {
				throw new CawNoRecordFoundException("No record in " + tableName + " found with row key " + rowkeys);
			}

			RowList responseRow = new RowList();
			responseRow.setRows(new ArrayList<Row>());

			if (columns != null) {
				responseRow.setFamilyNames(ConvertUtil.filterQualifiers(columns));
			} else {
				responseRow.setFamilyNames(pool.getTableFamilies(tableName));
			}

			for (Result r : results) {
				if (!r.isEmpty()) {
					Row row = ConvertUtil.convert(r);
					responseRow.getRows().add(row);
				}
			}

			if (responseRow.getRows().size() == 0) {
				throw new CawNoRecordFoundException("No record in " + tableName + " found with row keys " + rowkeys);
			}

			return responseRow;
		} catch (IOException e) {
			throw new CawException("Failed to get. Reason: " + e.getMessage(), e);
		} finally {
			if (table != null) {
				pool.putTable(table);
			}
		}

	}

	private Row executeGet(String tableName, final String rowkey, List<String> columns, Integer maxversions, Long timestamp) throws CawException {
		List<String> rowkeys = new ArrayList<String>();
		rowkeys.add(rowkey);

		RowList rowList = executeGets(tableName, rowkeys, columns, maxversions, timestamp);
		Row row = rowList.getRows().get(0);
		row.setFamilyNames(rowList.getFamilyNames());
		return row;
	}

	@Override
	public Row get(String tableName, final String rowkey, List<String> columns) throws CawException {
		return executeGet(tableName, rowkey, columns, 1, null);
	}

	@Override
	public Row get(String tableName, final String rowkey, List<String> columns, Long version) throws CawException {
		return executeGet(tableName, rowkey, columns, 1, version);
	}

	@Override
	public Row get(String tableName, String rowkey, List<String> columns, Integer numberOfVersions) throws CawException {
		return executeGet(tableName, rowkey, columns, numberOfVersions, null);
	}

	@Override
	public RowList multiget(String tableName, List<String> rowkeys, List<String> columns) throws CawException {
		return executeGets(tableName, rowkeys, columns, 1, null);
	}

	@Override
	public RowList multiget(String tableName, List<String> rowkeys, List<String> columns, Long version) throws CawException {
		return executeGets(tableName, rowkeys, columns, 1, version);
	}

	@Override
	public RowList multiget(String tableName, List<String> rowkeys, List<String> columns, Integer numberOfVersions) throws CawException {
		return executeGets(tableName, rowkeys, columns, numberOfVersions, null);
	}

	public HBaseTablePool getPool() {
		return pool;
	}

	public void setPool(HBaseTablePool pool) {
		this.pool = pool;
	}
}
